package com.JadePenG.demo;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 接收上游发送的单词, 进行统计, 打印统计结果
 */
public class WordCountBolt extends BaseRichBolt {
    //private HashMap<String, Integer> map = new HashMap<>();
    private static ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<String, Integer>();

    /**
     * 初始化变量  可以写在该方法中, 也可以写在方法外 private
     *
     * @param map
     * @param topologyContext
     * @param outputCollector
     */
    @Override
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {

    }

    /**
     * 接收上游splitBolt发送的单词, 将单词进行次数的统计, 将统计结果打印
     *
     * @param tuple
     */
    @Override
    public void execute(Tuple tuple) {
        //获取上游发送的数据
        String word = tuple.getStringByField("word");
        //Integer score = tuple.getIntegerByField("score"); 这个score始终是1  这个score可以不发
        //统计次数 每次单词发送过来该单词的次数 +1, 但是第一次不用
        //当前的次不是第一次统计  +1
        if (map.get(word) != null && map.get(word) > 0) {
            map.put(word, map.get(word) + 1);
        } else {//第一次统计该单词 不加
            map.put(word, 1);
        }
        //上面是统计的动作，每次过来一个单词都会统计，打印map
        System.out.println(map);
        //问题-->不需要打印每次结果
    }

    /**
     * 不需要再向下游发送, 可以不用再定义
     *
     * @param outputFieldsDeclarer
     */
    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

    }
}
